package com.etl.component.basic;

import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.ValidationColumnDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.types.Row;

import java.util.List;
import java.util.Map;

/**
 * 描述：
 *  数值校验
 * @author xianggj
 * @Date 2021/12/3 13:41
 **/
public class NumericalValidationComponent extends AbstractFlinkComponent {
    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<ValidationColumnDTO> validationColumnDTOS =
                (List<ValidationColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        SetData input = (SetData)datas.get(0).getValue();
        DataSet<Row> dataset = (DataSet<Row>) input.getDataset();
        List<TableColumn> columns = input.getColumns();
        setIndex(validationColumnDTOS, columns);
        DataSet<Row> result = dataset.filter(e->{
            for (ValidationColumnDTO validationColumnDTO  : validationColumnDTOS) {
                int index = validationColumnDTO.getIndex();
                Object field = e.getField(index);
                Boolean judge = validationColumnDTO.getFunction()
                        .judge(field, null);
                if (judge) {
                    return false;
                }
            }
            return true;
        });
        return execute(result, columns,"执行数值校验操作");
    }


    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        ColumnData input = (ColumnData)datas.get(0).getValue();
        List<TableColumn> columns = input.getColumns();
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.NUMERICAL_VALIDATION;
    }
}
